#include "udp_endpoint.h"
#include <chrono>

struct UdpEndpoint::MsgWrapper
{
    std::string msg_; asio::ip::udp::endpoint end_point_;
    MsgWrapper(
        const std::string& msg, const asio::ip::udp::endpoint& end_point) :msg_(msg), end_point_(end_point) { }
};

 UdpEndpoint::UdpEndpoint(asio::io_service& io_service, const char* local_addr, int local_port)
    :io_service_(io_service),
    socket_(io_service_),
    send_pk_timer_(io_service),
    rateKBs_(0),
    send_internal_msecs_(0)
{ 
    asio::ip::udp::endpoint listen_endpoint(asio::ip::address::from_string(local_addr), local_port);
    socket_.open(listen_endpoint.protocol());
    socket_.set_option(asio::ip::udp::socket::reuse_address(true));
    socket_.bind(listen_endpoint);
}

UdpEndpoint::~UdpEndpoint() { socket_.close(); }

void UdpEndpoint::start()
{
    if (send_internal_msecs_ > 0)
        setSendPkTimer();

    do_asyn_receive();
}

void UdpEndpoint::sendto(const std::string& msg, const char* remote_addr, int remote_port)
{
    SpMsgWrapper spmsg = std::make_shared<MsgWrapper>(msg, 
        asio::ip::udp::endpoint(asio::ip::address::from_string(remote_addr), remote_port));
    io_service_.post(std::bind(&UdpEndpoint::sendtoMessageInloop, shared_from_this(), spmsg));
}

void UdpEndpoint::sendto(const std::string& msg, const asio::ip::udp::endpoint& endpoint)
{
    SpMsgWrapper spmsg = std::make_shared<MsgWrapper>(msg, endpoint);
    io_service_.post(std::bind(&UdpEndpoint::sendtoMessageInloop, shared_from_this(), spmsg));
}

void UdpEndpoint::setMessageCallback(const MessageCallback& cb) { msg_callback_ = cb; }

void UdpEndpoint::join_multicast_group(const char* addr)
{
    socket_.set_option(
        asio::ip::multicast::join_group(asio::ip::address::from_string(addr)));
}

void UdpEndpoint::leave_multicast_group(const char* addr)
{
    socket_.set_option(
        asio::ip::multicast::leave_group(asio::ip::address::from_string(addr)));
}

void UdpEndpoint::setEncodeBuffer(const std::shared_ptr<MessageEncodeBuffer>& encodebuffer)
{
    msg_encode_buffer_ = encodebuffer;
}

void UdpEndpoint::setDecodeBuffer(const std::shared_ptr<MessageDecodeBuffer>& decodebuffer)
{
    msg_decode_buffer_ = decodebuffer;
}

void UdpEndpoint::do_asyn_receive()
{
    socket_.async_receive_from(asio::buffer(data_, max_length), sender_endpoint_, std::bind(
        &UdpEndpoint::handle_receive_from, shared_from_this(), std::placeholders::_1, std::placeholders::_2));
}

void UdpEndpoint::sendtoMessageInloop(const SpMsgWrapper& msgptr)
{
    bool is_writing = !msg_list_.empty();
    
    if (msg_encode_buffer_)
    {
        std::string& msg = msgptr->msg_;
        FecPacketList sendpackets = msg_encode_buffer_->appendAndTryEncode(msg.c_str(), msg.size());
        if (!sendpackets.empty())
        {
            std::list<SpMsgWrapper> send_msgs;
            for (auto& packet : sendpackets)
            {
                SpMsgWrapper spmsg = 
                    std::make_shared<MsgWrapper>(std::string(packet->get_packet_payload(), packet->get_packet_length()), msgptr->end_point_);
                send_msgs.push_back(spmsg);
            }
            msg_list_.insert(msg_list_.end(), send_msgs.begin(), send_msgs.end());
        }
    }
    else
        msg_list_.push_back(msgptr);

    if (send_internal_msecs_==0 && !is_writing && !msg_list_.empty())
    {
        sendToOneMessageInLoop();
    }
}

void UdpEndpoint::sendToOneMessageInLoop()
{
    assert(!msg_list_.empty());
    SpMsgWrapper send_msg = msg_list_.front();
    
    socket_.async_send_to(asio::buffer(send_msg->msg_.c_str(), send_msg->msg_.size()), send_msg->end_point_, std::bind(
        &UdpEndpoint::handle_sendto_complete, shared_from_this(), std::placeholders::_1, std::placeholders::_2));
}

void UdpEndpoint::handle_sendto_complete(const asio::error_code& error, size_t /*bytes_sent*/)
{
    if (error)
    {
        //LOG_PRINT(MAIN, warning, "UdpEndpoint::handle_sendto_complete error {}", error.message());
    }
    msg_list_.pop_front();

    if (send_internal_msecs_==0 && !msg_list_.empty())
    {
        sendToOneMessageInLoop();
    }
}

void UdpEndpoint::handle_receive_from(const asio::error_code& error, size_t bytes_recvd)
{
    do 
    {
        if (error || bytes_recvd <= 0)
        {
            //LOG_PRINT(MAIN, info, "UdpEndpoint::handle_receive_from error {}", error.value());
            break;
        }

        if (msg_decode_buffer_)
        {
            MessageList msgs = msg_decode_buffer_->appendAndTryDecode(data_, bytes_recvd);
            for (auto& msg : msgs)
            {
                msg_callback_(shared_from_this(), sender_endpoint_, msg.c_str(), msg.size());
            }
        }
        else
        {
            msg_callback_(shared_from_this(), sender_endpoint_, data_, bytes_recvd);
        }
    } while (false);

    do_asyn_receive();
}


void UdpEndpoint::setSendPkTimer()
{
    assert(send_internal_msecs_ > 0);
    send_pk_timer_.expires_from_now(std::chrono::milliseconds(send_internal_msecs_));
    send_pk_timer_.async_wait(std::bind(&UdpEndpoint::handleSendPkTimter, this));
}

void UdpEndpoint::handleSendPkTimter()
{
    setSendPkTimer();
    if (msg_list_.empty())
        return;
    sendToOneMessageInLoop();
}